package com.issac.messagequeue.mqserver.datacenter;

import com.issac.messagequeue.mqserver.common.MqException;
import com.issac.messagequeue.mqserver.coreclass.Binding;
import com.issac.messagequeue.mqserver.coreclass.Exchange;
import com.issac.messagequeue.mqserver.coreclass.MSGQueue;
import com.issac.messagequeue.mqserver.coreclass.Message;
import lombok.Data;
import lombok.extern.slf4j.Slf4j;

import java.io.IOException;
import java.util.LinkedList;
import java.util.List;
import java.util.concurrent.ConcurrentHashMap;

@Slf4j
@Data
public class MemoryDataCenter {
    private ConcurrentHashMap<String, Exchange> exchangeMap = new ConcurrentHashMap<>();
    private ConcurrentHashMap<String, MSGQueue> msgQueueMap = new ConcurrentHashMap<>();
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Binding>> bindingsMap = new ConcurrentHashMap<>();
    private ConcurrentHashMap<String, Message> messageMap = new ConcurrentHashMap<String, Message>();
    private ConcurrentHashMap<String, LinkedList<Message>> queueMessageMap = new ConcurrentHashMap<>();
    private ConcurrentHashMap<String, ConcurrentHashMap<String, Message>> queueMessageWaitAckMap = new ConcurrentHashMap<>();

    // 针对Exchange的操作
    public void addExchange(Exchange exchange) {
        exchangeMap.put(exchange.getName(), exchange);
        log.info("[MemoryDataCenter] 新交换机添加成功！ exchangeName=" + exchange.getName());
    }

    public void deleteExchange(String exchangeName) {
        exchangeMap.remove(exchangeName);
        log.info("[MemoryDataCenter] 交换机删除成功！ exchangeName=" + exchangeName);
    }

    public Exchange getExchange(String exchangeName) {
        return exchangeMap.get(exchangeName);
    }

    // 针对MSGQueue的操作
    public void addMSGQueue(MSGQueue queue) {
        msgQueueMap.put(queue.getName(), queue);
        log.info("[MemoryDataCenter] 新消息队列添加成功！ msgQueueName=" + queue.getName());
    }

    public void deleteMsgQueue(String msgQueueName) {
        msgQueueMap.remove(msgQueueName);
        log.info("[MemoryDataCenter] 新消息队列删除成功！ msgQueueName=" + msgQueueName);
    }

    public MSGQueue getMsgQueue(String msgQueueName) {
        return msgQueueMap.get(msgQueueName);
    }

    // 针对Binding的操作
    public void addBingding(Binding binding) throws MqException {
        // ConcurrentHashMap<String, ConcurrentHashMap<String, Binding>> bindingsMap
        //                   bindingName，             bindingName
        // bindingMap<String, Binding>

        // 1. 检查hashmap是否存在，不存在-创建
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(),
                k -> new ConcurrentHashMap<>());
        // 2. 存在-add
        synchronized (bindingMap) {
            if (bindingMap.get(binding.getQueueName()) != null) {
                throw new MqException("[MemoryDataCenter] 绑定已存在！ enchangeName=" + binding.getExchangeName()
                        + ", queueName=" + binding.getQueueName());
            }
            bindingMap.put(binding.getQueueName(), binding);
        }
        log.info("[MemoryDataCenter] 新绑定添加成功！ exchangeName=" + binding.getExchangeName()
                + ", queueName=" + binding.getQueueName());
    }

    public void deleteBinding(Binding binding) throws MqException {
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(binding.getExchangeName());
        if (bindingMap == null) {
            throw new MqException("[MemoryDataCenter] 绑定不存在！ exchangeName=" + binding.getExchangeName()
                    + ", queueName=" + binding.getQueueName());
        }
        bindingMap.remove(binding.getQueueName());
        log.info("[MemoryDataCenter] 绑定删除成功！ exchangeName=" + binding.getExchangeName()
                + ", queueName=" + binding.getQueueName());
    }

    public Binding getSpecifiedBinding(String exchangeName, String queueName) {
        ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.get(exchangeName);
        if (bindingMap == null) {
            return null;
        }
        return bindingMap.get(queueName);
    }

    public ConcurrentHashMap<String, Binding> getBindings(String exchangeName) {
        return bindingsMap.get(exchangeName);
    }

    // 针对消息的操作
    public void addMessage(Message message) {
        messageMap.put(message.getMessageId(), message);
        log.info("[MemoryDataCenter] 新消息添加成功！ messageId+" + message.getMessageId());
    }

    public Message getMessageById(String messageId) {
        return messageMap.get(messageId);
    }

    public void deleteMessageById(String messageId) {
        messageMap.remove(messageId);
        log.info("[MemoryDataCenter] 消息移除成功！ messageId=" + messageId);
    }

    public void sendMessage(Message message, MSGQueue queue) {
        // 1. 根据queueName找到对应的消息List
        LinkedList<Message> messageList = queueMessageMap.computeIfAbsent(queue.getName(), k -> new LinkedList<>());
        // 2. message加入到queue对应的List
        synchronized (messageList) {
            messageList.add(message);
        }
        // 3. 在消息中心插入消息
        addMessage(message);
        log.info("[MemoryDataCenter] 消息被投递到队列中！ messageId=" + message.getMessageId() + ", queueName=" + queue.getName());
    }

    public Message getMessageFromQueue(String queueName) {
        // 实际上就是把message从List中头取出
        // 1. 根据queueName查找对应的MessageList
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if (messages == null) {
            return null;
        }
        // 2. 取消息
        synchronized (messages) {
            if (messages.size() == 0) {
                return null;
            }
            Message currentMessage = messages.remove(0);
            log.info("[MemoryDataCenter] 消息从队列中取走！ messageId=" + currentMessage.getMessageId() + ", queueName=" + queueName);
            return currentMessage;
        }
    }

    // 获取指定队列中的消息个数
    public int getMessageCount(String queueName) {
        LinkedList<Message> messages = queueMessageMap.get(queueName);
        if (messages == null) {
            return 0;
        }
        synchronized (messages) {
            return messages.size();
        }
    }

    // 关于未确认消息的操作
    public void addMessageWaitAck(String queueName, Message message) {
        ConcurrentHashMap<String, Message> messageHashMap = queueMessageWaitAckMap.computeIfAbsent(queueName,
                k -> new ConcurrentHashMap<>());
        messageHashMap.put(message.getMessageId(), message);
        log.info("[MemoryDataCenter] 消息添加到待确认队列！ messageId=" + message.getMessageId() + ", queueName=" + queueName);
    }

    public void deleteMessageWaitAck(String queueName, String messageId) throws MqException {
        ConcurrentHashMap<String, Message> messageHashMap = queueMessageWaitAckMap.get(queueName);
        if (messageHashMap == null) {
            throw new MqException("[MemoryDataCenter] 该待确认消息队列无消息！ queueName=" + queueName);
        }
        messageHashMap.remove(messageId);
        log.info("[MemoryDataCenter] 消息从待确认队列删除！ messageId=" + messageId + ", queueName=" + queueName);
    }

    public Message getMessageWaitAck(String queueName, String messageId) throws MqException {
        ConcurrentHashMap<String, Message> messageHashMap = queueMessageWaitAckMap.get(queueName);
        if (messageHashMap == null) {
            throw new MqException("[MemoryDataCenter] 该待确认消息队列无消息！ queueName=" + queueName);
        }
        return messageHashMap.get(messageId);
    }

    public void recovery(DiskDataCenter diskDataCenter) throws IOException, MqException, ClassNotFoundException {
        log.info("[MemoryDataCenter] recovery: 正在把硬盘数据恢复到内存中......");
        // 1. 清空之前的数据
        exchangeMap.clear();
        msgQueueMap.clear();
        bindingsMap.clear();
        messageMap.clear();
        queueMessageMap.clear();
        log.info("[MemoryDataCenter] recovery: 内存数据已清空！ ✔");
        // 2. 恢复交换机数据
        List<Exchange> exchangeList = diskDataCenter.searchAllExchanges();
        for (Exchange exchange : exchangeList) {
            exchangeMap.put(exchange.getName(), exchange);
        }
        log.info("[MemoryDataCenter] recovery: 交换机数据已恢复！ ✔");
        // 3. 恢复队列数据
        List<MSGQueue> msgQueueList = diskDataCenter.searchAllMSGQueues();
        for (MSGQueue msgQueue : msgQueueList) {
            msgQueueMap.put(msgQueue.getName(), msgQueue);
        }
        log.info("[MemoryDataCenter] recovery: 消息队列数据已恢复！ ✔");
        // 4. 恢复绑定数据
        List<Binding> bindingList = diskDataCenter.searchAllBindings();
        for (Binding binding : bindingList) {
            ConcurrentHashMap<String, Binding> bindingMap = bindingsMap.computeIfAbsent(binding.getExchangeName(),
                    k -> new ConcurrentHashMap<>());
            bindingMap.put(binding.getQueueName(), binding);
        }
        log.info("[MemoryDataCenter] recovery: 绑定数据已恢复！ ✔");
        // 5. 恢复消息数据
        for (MSGQueue msgQueue : msgQueueList) {
            LinkedList<Message> messageList = diskDataCenter.loadAllMessageFromQueue(msgQueue.getName());
            queueMessageMap.put(msgQueue.getName(), messageList);
            for (Message message : messageList) {
                messageMap.put(message.getMessageId(), message);
            }
        }
        log.info("[MemoryDataCenter] recovery: 消息数据已恢复！ ✔");
        log.info("[MemoryDataCenter] recovery: 硬盘数据恢复到内存中已完成！");
    }
}
